package org.example.com.atguigu.day06;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.catalyst.expressions.Second;
import org.apache.spark.streaming.Seconds;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Iterator;

public class Window {
    public static void main(String[] args) throws InterruptedException, URISyntaxException, IOException {
        // second 5 : 五秒一个批次
        JavaStreamingContext ssc = new JavaStreamingContext(new SparkConf().setMaster("local[*]").setAppName("test"), Seconds.apply(5));
        ssc.sparkContext().setLogLevel("error");  // 只展示报错信息,不看别的日志信息,方便查看运行结果
        JavaReceiverInputDStream<String> ds = ssc.socketTextStream("hadoop102", 9999);

        JavaPairDStream<String, Integer> ds2 = ds.flatMapToPair(new PairFlatMapFunction<String, String, Integer>() {
            @Override
            public Iterator<Tuple2<String, Integer>> call(String s) throws Exception {
                String[] arr = s.split(" ");
                ArrayList<Tuple2<String, Integer>> res = new ArrayList<>();
                for (String wc : arr) {
                    res.add(new Tuple2<>(wc, 1));
                }
                return res.iterator();
            }
        });

        // 下面ds2需要复用,使用缓存
        ds2.cache();

        // 打印ds2的数据 方便验证下面的窗口是否正确
        JavaPairDStream<String, Integer> ds4 = ds2.mapToPair(new PairFunction<Tuple2<String, Integer>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<String, Integer> t2) throws Exception {
                return new Tuple2<>("---->" + t2._1, t2._2);
            }
        });
        // 一个批次的结果打印
        ds4.print();

        // TODO 窗口长度和滑动长度必须是批次时间的整数倍
        /*// 窗口大小15秒,滑动长度5秒
        JavaPairDStream<String, Integer> ds21 = ds2.window(Seconds.apply(15), Seconds.apply(5));

        JavaPairDStream<String, Integer> ds3 = ds21.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });*/
        // 上面写法,有下面这个更简洁的写法  reduceByKeyAndWindow = window + reduceByKey
        JavaPairDStream<String, Integer> ds3 = ds2.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        }, Seconds.apply(15), Seconds.apply(5));

        // 一个窗口的结果打印
        ds3.print();

        // 启动
        ssc.start();
        // TODO 如何通过外部给程序信号让程序停止?
        // TODO 方法1: 在程序中监听mysql某个表的某个字段,如果字段值改变,代表我们需要停止程序
        // TODO 方法2: 在程序中监听mysql某个目录是否被删除,如果被删除代表我们需要停止程序
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop102:8020"), new Configuration());
        // 如果目录存在,则一直循环;
        while (fileSystem.exists(new Path("hdfs://hadoop102:8020/switch"))){
            // 每隔一段时间查看目录是否存在
            Thread.sleep(5000);
        }
        // TODO 如何让接受的数据处理完成之后停止,只会要让stop中的stopGracefully=true即可;
        // 目录被删,while退出,执行stop
        ssc.stop(true,true);
        // 阻塞  // 阻塞不需要,因为while起到了阻塞的效果
        // ssc.awaitTermination();
    }
}
